package com.xiaochong.tir.apiservice.config.hbase;



import com.xiaochong.tir.common.exception.hbase.HbaseCloseExcetion;
import com.xiaochong.tir.common.exception.hbase.HbaseConnectionExcetion;
import com.xiaochong.tir.common.exception.hbase.HbaseGetTableExcetion;
import com.xiaochong.tir.common.exception.hbase.HbaseTablePutExcetion;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.io.IOException;
import java.util.List;

/** Hbase 连接
 * Created by cx on 2018/4/24.
 */
@Component
public class HbaseConnectionFactory {

    //master 连接配置
    private static Configuration configuration;

    private static Connection connection;

    @Value("${hbase.quorum}")
    private String quorum;
    @Value("${hbase.clientPort}")
    private String clientPort;
    @Value("${hbase.rpc.timeout}")
    private String timeout;
    @Value("${hbase.number}")
    private String number;
    @Value("${hbase.pause}")
    private String pause;
    @Value("${zookeeper.retry}")
    private String retry;
    @Value("${zookeeper.intervalmill}")
    private String intervalmill;
    @Value("${hbase.operation.timeout}")
    private String operationTimeout;

    @PostConstruct
    private void init(){
        configuration = new Configuration();
        configuration.set("hbase.zookeeper.quorum",quorum);
        configuration.set("hbase.zookeeper.property.clientPort",clientPort);
        //rpc请求的超时时间默认为60秒,修改为5秒
        configuration.set("hbase.rpc.timeout", "5000");
        //客户端重试最大次数默认35,修改为3
        configuration.set("hbase.client.retries.number", number);
        //失败重试时等待时间
        configuration.set("hbase.client.pause", pause);
        //zookeeper重试次数修改为3
        configuration.set("zookeeper.recovery.retry", retry);
        //zookeeper重试的休眠时间,默认为1秒,修改为200毫秒
        configuration.set("zookeeper.recovery.retry.intervalmill", intervalmill);
        //HBase客户端发起一次数据操作直至得到响应之间总的超时时间默认为20分钟,修改为30秒
        configuration.set("hbase.client.operation.timeout", "30000");
        try {
            connection = ConnectionFactory.createConnection(configuration);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    /**
     * 得到connection连接
     * @return
     * @throws IOException
     * @throws HbaseConnectionExcetion
     */
    public  Connection getConnection() throws HbaseConnectionExcetion {
        try {
            if(null == connection){
                connection = ConnectionFactory.createConnection(configuration);
            }

        } catch (IOException e) {
            throw new  HbaseConnectionExcetion("Hbase 获取Connection连接失败!", e);

        }

        return  connection;
    }


    /**
     * 获取table
     * @param tableName 表名
     * @param conn
     * @return
     * @throws HbaseConnectionExcetion Connection连接异常
     * @throws HbaseGetTableExcetion getTable时出现Io异常
     */
    public Table getTable(Connection conn, String tableName) throws HbaseConnectionExcetion, HbaseGetTableExcetion {
        Table table;
        try {
            table = conn.getTable(TableName.valueOf(tableName));
        } catch (IOException e) {
            throw new HbaseGetTableExcetion("hbase getTable时出现Io异常", e);
        }
        return table;
    }

    /**
     * 单个添加
     * @param table
     * @param put
     * @throws HbaseTablePutExcetion
     */
    public  void tablePut(Table table,Put put) throws HbaseTablePutExcetion {
        try {
            table.put(put);
        } catch (IOException e) {
            // TODO Auto-generated catch block
            throw new HbaseTablePutExcetion("hbase 像表中添加数据时,出现异常", e);
        }
    }

    /**
     *  批量添加
     * @param table
     * @param putList
     * @throws HbaseTablePutExcetion
     */
    public  void tablePut(Table table,List<Put> putList) throws HbaseTablePutExcetion {
        try {
            table.put(putList);
        } catch (IOException e) {
            // TODO Auto-generated catch block
            throw new HbaseTablePutExcetion("hbase 像表中添加数据时,出现异常", e);
        }
    }

    /**
     * hbase 关闭连接
     * @param conn
     * @param table
     * @throws HbaseCloseExcetion
     */
    public  void closeAll(Connection conn, Table table) throws HbaseCloseExcetion {
        try {
            table.close();
        } catch (IOException e) {
            throw new HbaseCloseExcetion("hbase 关闭异常", e);
        }

    }



}
